package cn.com.lyb.flink.transfrom;

import cn.com.lyb.flink.bean.WaterSensor;
import cn.com.lyb.flink.udf.FilterFunctionDemo;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * filter转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤条件，
 * 对于每一个流内元素进行判断，若为true则元素正常输出，若为false则元素被过滤掉。
 * 进行filter转换之后的新数据流的数据类型与原数据流是相同的。
 * filter转换需要传入的参数需要实现FilterFunction接口，
 * 而FilterFunction内要实现filter()方法，就相当于一个返回布尔类型的条件表达式。
 *
 * 案例需求：下面的代码会将数据流中传感器id为sensor_1的数据过滤出来。
 */
public class FilterDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 3),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3));

//        sensorDS.filter(new FilterFunction<WaterSensor>() {
//            @Override
//            public boolean filter(WaterSensor waterSensor) throws Exception {
//                return waterSensor.getId().equals("s2");
//            }
//        }).print();

        sensorDS.filter(new FilterFunctionDemo("s2")).print();

        env.execute();
    }
}
